package com.iot.service.mqtt.server;


import com.iot.service.mqtt.codec.MqttWebSocketCodec;
import com.iot.service.mqtt.codec.TlMqttMessageCodec;
import com.iot.service.mqtt.codec.decoder.*;
import com.iot.service.mqtt.codec.encoder.*;
import com.iot.service.mqtt.config.TlMqttProperties;
import com.iot.service.mqtt.config.TlPortProperties;
import com.iot.service.mqtt.config.TlSslProperties;
import com.iot.service.mqtt.handler.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslProvider;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.InputStream;

/**
 * @Author: hszhou
 * @Date: 2024/11/15 15:10
 * @Description: 启动类
 */
@Component
@RequiredArgsConstructor
public class TlMqttServer {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttServer.class);


    private final TlMqttConnectHandler mqttConnectHandler;

    private final TlMqttPublishHandler mqttPublishHandler;

    private final TlMqttSubScribeHandler mqttSubScribeHandler ;

    private final TlMqttUnSubScribeHandler mqttUnSubScribeHandler ;

    private final TlMqttHeartBeatHandler mqttHeartBeatHandler ;

    private final TlMqttDisconnectHandler mqttDisconnectHandler;

    private final TlMqttPubCompHandler mqttPubCompHandler;

    private final TlMqttPubAckHandler mqttPubAckHandler;

    private final TlMqttPubRecHandler mqttPubRecHandler ;

    private final TlMqttPubRelHandler mqttPubRelHandler ;

    private final TlMqttConnackEncoder mqttConnackEncoder ;

    private final TlMqttHeaderBeatEncoder mqttHeaderBeatEncoder;

    private final TlMqttSubAckEncoder mqttSubAckEncoder ;

    private final TlMqttUnSubAckEncoder mqttUnSubAckEncoder;

    private final TlMqttPublishEncoder mqttPublishEncoder;

    private final TlMqttPubAckEncoder mqttPubAckEncoder;

    private final TlMqttPubRecEncoder mqttPubRecEncoder;

    private final TlMqttPubRelEncoder mqttPubRelEncoder;

    private final  TlMqttPubCompEncoder mqttPubCompEncoder;

    private final TlMqttExceptionHandler mqttExceptionHandler ;

    private final TlHeartbeatEventTriggered tlHeartbeatEventTriggered;

    private final  TlMqttConnectDecoder connectDecoder;

    private final  TlMqttDisConnectDecoder disConnectDecoder;

    private final  TlMqttHeartBeatDecoder heartBeatDecoder;

    private final  TlMqttPubAckDecoder pubAckDecoder;

    private final  TlMqttPubCompDecoder pubCompDecoder;

    private final  TlMqttPublishDecoder publishDecoder;

    private final  TlMqttPubRecDecoder pubRecDecoder;

    private final  TlMqttPubRelDecoder pubRelDecoder;

    private final  TlMqttSubscribeDecoder subscribeDecoder;

    private final  TlMqttUnSubscribeDecoder unSubscribeDecoder;

    private  final TlMqttProperties mqttConfig;


    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private SslContext sslContext;

    @PostConstruct
    public void start(){
      try{
          TlPortProperties portConfig = mqttConfig.getPort();
          TlSslProperties sslProperties = mqttConfig.getSsl();
          bossGroup = new NioEventLoopGroup();
          workerGroup = new NioEventLoopGroup();
          if(sslProperties.isEnabled()){
              sslContext=sslContext();
          }
          mqttServer(portConfig,sslProperties.isEnabled());
         websocket(portConfig,sslProperties.isEnabled());
      }catch (Exception e){
          logger.error("启动失败【{}】",e.getMessage());
      }
    }

    public void  websocket(TlPortProperties port,boolean ssl) throws InterruptedException {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                // handler在初始化时就会执行
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        // Netty提供的心跳检测
                        pipeline.addFirst("idle", new IdleStateHandler(0, 0, 10));
                        // 将请求和应答消息编码或解码为HTTP消息
                        pipeline.addLast("http-codec", new HttpServerCodec());
                        // 将HTTP消息的多个部分合成一条完整的HTTP消息
                        pipeline.addLast("aggregator", new HttpObjectAggregator(1048576));
                        // 将HTTP消息进行压缩编码
                        pipeline.addLast("compressor ", new HttpContentCompressor());
                        pipeline.addLast("protocol", new WebSocketServerProtocolHandler("/mqtt", "mqtt,mqttv3.1,mqttv3.1.1", true, 65536));
                        pipeline.addLast("mqttWebSocket", new MqttWebSocketCodec());
                        addMqttHandler(socketChannel, pipeline, ssl);
                    }
                });

        ChannelFuture channelFuture =null;
        if(ssl){
            channelFuture =  bootstrap.bind(port.getSslWebsocket()).sync();
        }else{
            channelFuture = bootstrap.bind(port.getWebsocket()).sync();
        }

        logger.info("mqtt websocket server start success port is {}",port.getWebsocket());
        channelFuture.channel().closeFuture().addListener(r->{
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        });
    }



    public void mqttServer(TlPortProperties port,boolean ssl){

        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap.group(bossGroup,workerGroup)
             .option(ChannelOption.SO_BACKLOG,1024)
             .option(ChannelOption.SO_REUSEADDR,true)
             .childOption(ChannelOption.SO_KEEPALIVE,true)
             .childOption(ChannelOption.TCP_NODELAY,true)
             .childOption(ChannelOption.SO_SNDBUF,32*1024)
             .childOption(ChannelOption.SO_RCVBUF,32*1024)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    addMqttHandler(socketChannel, pipeline, ssl);
                 }
            });

            ChannelFuture channelFuture =null;
            if(ssl){
                channelFuture =  bootstrap.bind(port.getSslMqtt()).sync();
            }else{
                 channelFuture = bootstrap.bind(port.getMqtt()).sync();
            }

            logger.info("mqtt server start success port is {}",port.getMqtt());
            channelFuture.channel().closeFuture().addListener(r->{
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            });
        }catch (Exception e){
            logger.error("error 【{}】",e.getMessage());
        }
    }


    private void addMqttHandler(SocketChannel socketChannel, ChannelPipeline pipeline, boolean ssl) {
        if(ssl){
            pipeline.addFirst("ssl",sslContext.newHandler(socketChannel.alloc()));
        }
        //协议解析
        pipeline.addLast(new TlMqttMessageCodec(connectDecoder,
                        disConnectDecoder,heartBeatDecoder,pubAckDecoder,
                        pubCompDecoder,publishDecoder,pubRecDecoder,
                        pubRelDecoder,subscribeDecoder,unSubscribeDecoder))
                // pipeline.addLast(tlMqttMessageCodec)
                //连接处理
                .addLast(mqttConnectHandler)
                //publish消息接收处理器
                .addLast(mqttPublishHandler)
                //订阅主题处理
                .addLast(mqttSubScribeHandler)
                //取消订阅处理
                .addLast(mqttUnSubScribeHandler)
                //心跳检测处理
                .addLast(mqttHeartBeatHandler)
                //断开连接处理
                .addLast(mqttDisconnectHandler)
                //pubcomp处理
                .addLast(mqttPubCompHandler)
                //puback处理
                .addLast(mqttPubAckHandler)
                //rec处理
                .addLast(mqttPubRecHandler)
                //rel处理
                .addLast(mqttPubRelHandler)
                //ack编码
                .addLast(mqttConnackEncoder)
                //心跳恢复编码
                .addLast(mqttHeaderBeatEncoder)
                //订阅ack编码
                .addLast(mqttSubAckEncoder)
                //取消订阅ack编码
                .addLast(mqttUnSubAckEncoder)
                //publish编码
                .addLast(mqttPublishEncoder)
                //puback编码
                .addLast(mqttPubAckEncoder)
                //订阅ack编码
                //rec编码
                .addLast(mqttPubRecEncoder)
                .addLast(mqttPubRelEncoder)
                //comp编码
                .addLast(mqttPubCompEncoder)
                //异常处理
                .addLast(mqttExceptionHandler)
                //心跳检测
                .addLast(tlHeartbeatEventTriggered)
                //心跳
                .addLast( "idleHandler", new IdleStateHandler(
                        0, 0,  10));
    }

    /**
     * 构建证书
     * @return SslContext
     * @throws IOException 异常
     */
    public SslContext sslContext() throws IOException {
        // 使用 ClassPathResource 来加载资源文件
        ClassPathResource certResource = new ClassPathResource(mqttConfig.getSsl().getCertPath());
        ClassPathResource keyResource = new ClassPathResource(mqttConfig.getSsl().getPrivatePath());

        try (InputStream certStream = certResource.getInputStream();
             InputStream keyStream = keyResource.getInputStream()) {

            // 使用 SslContextBuilder 创建 SslContext，并传递 InputStream
            return SslContextBuilder.forServer(certStream, keyStream)
                    // 使用 OpenSSL 提供更好的性能 目前使用的是jdk自带的
                    .sslProvider(SslProvider.JDK)
                    .build();
        }
    }
}
